package com.intmall.flink.operator;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

public class TransformMapTest {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        // 传入MapFunction的实现类
        DataStreamSource<Event> streamSource = env.fromElements(new Event("Mary", "./home", 1000L),
                new Event("Bob", "./cart", 2000L));
        SingleOutputStreamOperator<String> result1 = streamSource.map(new MyMapper());
        result1.print("result1");

        // 传入匿名类，实现MapFunction
        SingleOutputStreamOperator<String> result2 = streamSource.map(new MapFunction<Event, String>() {
            @Override
            public String map(Event value) throws Exception {
                return value.getUser();
            }
        });
        result2.print("result2");

        // Lambda表达式
        SingleOutputStreamOperator<String> result3 = streamSource.map(value -> value.getUser());
        result3.print("result3");

        env.execute();

    }

    public static class MyMapper implements MapFunction<Event, String> {

        @Override
        public String map(Event value) throws Exception {
            return value.getUser();
        }
    }
}
